package cn.icast.nio.c4;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

import static cn.icast.nio.c2.ByteBufferUtil.debugAll;

@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        //创建线程名称
        Thread.currentThread().setName("boss");
        //建立服务器ssc
        ServerSocketChannel ssc = ServerSocketChannel.open();
        //给ssc服务器配置阻塞模式
        ssc.configureBlocking(false);
        //创建Selector选择器
        Selector boss = Selector.open();
        //把服务器boss注册到ssc服务器的SelectionKey上
        SelectionKey bossKey = ssc.register(boss, 0, null);
        //添加关注客户端连接事件
        bossKey.interestOps(SelectionKey.OP_ACCEPT);
        //给服务器设置绑定端口
        ssc.bind(new InetSocketAddress(8080));
      //  System.out.println("CPU的核心数"+Runtime.getRuntime().availableProcessors());
        //1.创建固定数量的worker  并初始化
        // 创建worker线程池

        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
        //2.创建线程池名称
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-" + i);
        }
            //1.创建固定数量的worker  并初始化
          //  Worker worker = new Worker("worker-0");
          //创建计数器
        AtomicInteger index = new AtomicInteger();
        while (true) {
            //boss选择器接受消息事件，有事件才激活，无事件处于阻塞状态,此处为监听事件
            boss.select();
            //处理事件的selectedKeys集合
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();

            //使用迭代器拿个每一个消息事件
            while (iter.hasNext()) {

                try {
                    //key就是一个消息事件
                    SelectionKey key = iter.next();
                    //删除掉处理过的消息事件
                    iter.remove();
                    //判断我们是要处理客户端连接问题，还是其他的：如可读、可写事件，下面是处理连接事件
                    if (key.isAcceptable()) {
                        //创建一个连接通道
                        SocketChannel sc = ssc.accept();
                        //取消客户端连接通道阻塞模式，改为不阻塞
                        sc.configureBlocking(false);

                        log.debug("connected...{}", sc.getRemoteAddress());//打印远程地址

                        log.debug("before register...{}", sc.getRemoteAddress());//打印远程地址
                        //初始化worker
                        //按顺序轮询 线程
                        log.debug("index...{}",index.getAndIncrement());
                        log.debug("worksLength...{}",workers.length);

                        workers[index.getAndIncrement() % workers.length].register(sc);//boss调用，初始化selector，启动worker-0
                        log.debug("after register...{}", sc.getRemoteAddress());//打印远程地址


                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);

                }
            }
        }
    }

    static class Worker implements Runnable {
        // 线程
        private Thread thread;
        //channel通道管理器
        private Selector selector;
        //添加消息队列
        private ConcurrentLinkedDeque<Runnable> queue = new ConcurrentLinkedDeque<>();
        //线程名称
        private String name;
        //控制线程开启，看线程是否开启过,start就是想让一个selector只用一个线程
        //volatile使其保持可见性
        //此处boolean变量用作标记使用
        private volatile boolean start = false;//还未初始化


        // 按fn+alt+insert组合建
        // 设置线程的名称


        public Worker(String name) {
            this.name = name;
        }

        //初始化线程和selector
        public void register(SocketChannel sc) throws IOException {
            // start为true时不创建线程，为false时才创建
            if (!start) {
                //将来执行selector方法，启动线程后this,就会执行run方法
                thread = new Thread(this, name);

                //创建selector
                selector = Selector.open();

                //线程启动
                thread.start();
                start = true;
            }

        /* queue.add(new Runnable() {
             @Override
             public void run() {
                 try {
                     sc.register(selector, SelectionKey.OP_READ, null);
                 } catch (ClosedChannelException e) {
                     throw new RuntimeException(e);
                 }
             }
         });*/
            queue.add(() -> {
                try {
                    //2.把worker上的selector关联 SocketChannle sc上的 selector,并关注可读事件
                    sc.register(selector, SelectionKey.OP_READ, null);
                } catch (ClosedChannelException e) {
                    throw new RuntimeException(e);
                }
            });
            selector.wakeup();//唤醒select方法
        }

        @Override
        public void run() {
            while (true) {
                try {
                    //开始监测事件
                    selector.select();//执行worker-0
                    //它取到的任务有可能是空的，此处我们做非空判断
                    Runnable task = queue.poll();

                    if (task != null) {
                        task.run();//在此处位置执行了 sc.register(selector, SelectionKey.OP_READ, null);
                    }

                    //如果有事件发生了，我就用selectedKeys得到所有事件
                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();

                    while (iter.hasNext()) {
                        //key事件
                        SelectionKey key = iter.next();
                        //删除已处理过的事件
                        iter.remove();
                        //判断处理哪一种事件,此处处理可读事件

                        if (key.isReadable()) {
                            //创建缓冲字节，有2G
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            //把buffer切换成读模式
                            buffer.flip();
                            // 创建SocketChannel通道，从key里面获取
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.debug("read...{}", channel.getRemoteAddress());//打印远程地址
                            // 从channel读取内容放进buffer
                            int read = channel.read(buffer);
                            if (read < 0) {
                                key.cancel();
                                break;
                            }

                            //打印结果
                            debugAll(buffer);
                        }
                    }


                } catch (IOException e) {
                    throw new RuntimeException(e);

                }

            }
        }
    }
}
